/**
 * Copyright 2013-2015
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.terramagnet.learn.rabbitmq.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import javax.annotation.Resource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.terramagnet.learn.rabbitmq.consumer.framework.BatchedAsynchrony;

/**
 *
 * @author lip
 */
@Service
public class Consumer extends BatchedAsynchrony<MessageSample> {

    private final ObjectMapper jsonUtil = new ObjectMapper();
    @Resource
    private JdbcTemplate jdbcTemplate;

    private String tableName = "message2";

    private int recieveCount;
    private int consumeCount;

    public void onRecieve(String message) {
        try {
            MessageSample p = jsonUtil.readValue(message, MessageSample.class);
            put(p);
            recieveCount++;
        } catch (IOException ex) {
            logger.error("json 反序列化失败", ex);
        }
    }

    @Override
    protected void take(Collection<MessageSample> data) {
        StringBuilder sql = new StringBuilder();
        List params = new ArrayList();
        sql.append("insert into ").append(tableName).append(" (id,message) values ");
        for (MessageSample ms : data) {
            sql.append("(?,?),");
            params.add(ms.getId());
            params.add(ms.getMessage());
        }
        sql.deleteCharAt(sql.length() - 1);

        jdbcTemplate.update(sql.toString(), params.toArray());
        consumeCount += data.size();
    }

    @Override
    protected String getThreadName() {
        return "thread-consumer";
    }

    public void setTableName(String tableName) {
        this.tableName = tableName;
    }

    @Scheduled(fixedRate = 5000)
    public void beat() {
        logger.info("I have recieved {} messages, and consumed {} works.", recieveCount, consumeCount);
    }

    @Override
    protected void afterLoad() {
        setBatchSize(50);
        recieveCount = deserialize("recieved");
        consumeCount = deserialize("consumed");
        super.afterLoad();
    }

    @Override
    protected void beforeShutdown() {
        super.beforeShutdown();
        addSerializable("recieved", recieveCount);
        addSerializable("consumed", consumeCount);
    }
}
